package com.tinem.platform.module.starter.jms.consumer.impl;

import com.alibaba.fastjson.JSONObject;
import com.tinem.platform.module.starter.jms.consumer.ConsumerService;
import com.tinem.platform.module.starter.jms.consumer.SubscribeListener;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jms.annotation.JmsListenerConfigurer;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerEndpointRegistrar;
import org.springframework.jms.config.JmsListenerEndpointRegistry;
import org.springframework.jms.config.SimpleJmsListenerEndpoint;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import org.springframework.stereotype.Service;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import javax.annotation.Resource;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import java.lang.reflect.Type;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author fengzhihao
 * @version v1
 * @program: platform
 * @className ActiveMqConsumerService
 * @description TODO
 * @site
 * @company
 * @create 2021/12/1 10:31:43
 */
@Slf4j
@Service
public class ActiveMqConsumerService<T> implements ConsumerService<T>, JmsListenerConfigurer {


    JmsListenerEndpointRegistrar jmsListenerEndpointRegistrar;
    @Resource
    JmsListenerContainerFactory topicListenerFactory;

    Map<String,MessageListener> messageListenerMap=new ConcurrentHashMap<>();
    @Override
    @SneakyThrows
    public String subscribeTopic(String topic, SubscribeListener<T> subscribeListener) {
        Type t = subscribeListener.getClass().getGenericInterfaces()[0];
        Type type = ((ParameterizedTypeImpl) t).getActualTypeArguments()[0];
        SimpleJmsListenerEndpoint simpleJmsListenerEndpoint = new SimpleJmsListenerEndpoint();
        simpleJmsListenerEndpoint.setId(UUID.randomUUID().toString());
        simpleJmsListenerEndpoint.setDestination(topic);
        MessageListener messageListener = new MessageListener() {
            @SneakyThrows
            @Override
            public void onMessage(Message m) {
                TextMessage textMessage = (TextMessage) m;
                log.info("call message:{},{}",textMessage.getJMSMessageID(),textMessage.getText());
                try {
                    subscribeListener.call(JSONObject.parseObject(textMessage.getText(), type));
                    log.info("call message success:{}",textMessage.getJMSMessageID());
                }catch (Exception e){
                    log.error("call message error:{}",textMessage.getJMSMessageID(),e);
                }
            }
        };
        messageListenerMap.put(simpleJmsListenerEndpoint.getId(),messageListener);
        simpleJmsListenerEndpoint.setMessageListener(messageListener);
        jmsListenerEndpointRegistrar.registerEndpoint(simpleJmsListenerEndpoint,topicListenerFactory);

        return simpleJmsListenerEndpoint.getId();
    }

    @Override
    public void cancelSubscribeTopic(Set<String> connectionIdSet) {
        JmsListenerEndpointRegistry jmsListenerEndpointRegistry = jmsListenerEndpointRegistrar.getEndpointRegistry();
        Collection<MessageListenerContainer> messageListenerContainers =  jmsListenerEndpointRegistry.getListenerContainers();
        for (MessageListenerContainer container : messageListenerContainers) {
            Object messageListener = ((DefaultMessageListenerContainer) container).getMessageListener();
            for (String connectionId : connectionIdSet){
                if(messageListenerMap.get(connectionId) == messageListener){
                    ((DefaultMessageListenerContainer) container).shutdown();
                }
            }
        }
    }

    @Override
    public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
        this.jmsListenerEndpointRegistrar = registrar;
    }
}
